package cn.fanyamin.group;

import java.io.Serializable;
import java.util.Date;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.JChannel;
import org.jgroups.blocks.NotificationBus;

import sun.misc.Signal;
import sun.misc.SignalHandler;




public class NotificationManager implements Runnable {
    private static final int MSG_INTERVAL_MS = 10000;
    /**
     * jgroups.udp.mcast_port
     */
    private static final String JGROUPS_UDP_MCAST_PORT = "jgroups.udp.mcast_port";

    /**
     * jgroups.udp.mcast_addr
     */
    private static final String JGROUPS_UDP_MCAST_ADDR = "jgroups.udp.mcast_addr";

    /** The Constant logger. */
    private static Log logger = LogFactory.getLog(NotificationManager.class);

    /** The Constant DEFAULT_CFG_FILE. */
    private static final String DEFAULT_CFG_FILE = "./etc/udp.xml";

    /** The start flag. */
    private volatile boolean startFlag = false;

    /** The stop request. */
    private volatile boolean stopRequest = false;

    /** The notification bus. */
    private NotificationBus notificationBus;

    private NotificationConsumer consumer;
    /** The my name. */
    private String myName;

    private String configFile = DEFAULT_CFG_FILE
            ;

    private Channel myChannel;
    
    /** The exector. */
    private ExecutorService exector = Executors.newSingleThreadScheduledExecutor();
 

    /**
     * Instantiates a new group manager.
     * 
     * @param busName
     *            the bus name
     * @param properties
     *            the properties
     * @param service
     *            the service
     * @throws Exception
     *             the exception
     */
    public NotificationManager(String mcastAddr, int mcastPort) throws Exception {
        System.setProperty(JGROUPS_UDP_MCAST_ADDR, mcastAddr);
        System.setProperty(JGROUPS_UDP_MCAST_PORT, "" + mcastPort);
        
        
    }

    /**
     * Send notification.
     * 
     * @param msg
     *            the msg
     */
    public void sendNotification(Serializable msg) {
        this.notificationBus.sendNotification(msg);
    }

    public void setName(String name) {
        this.myName = name;
    }
    
    /**
     * Clean.
     * 
     * @throws Exception
     *             error
     */
    public void clean() throws Exception {
       
    }

    /**
     * @see com.webex.mdp.service.IService#getName()
     * @return name
     */
    public String getName() {
        return this.myName;
    }

    /**
     * @see com.webex.mdp.service.IService#init()
     * @throws Exception
     *             error
     */
    public void init() throws Exception {
        this.myChannel = new JChannel(this.configFile);
        this.notificationBus = new NotificationBus(this.myChannel, this.myName);
        notificationBus.setConsumer(consumer);

    }

    /*
     * (non-Javadoc)
     * @see com.webex.mdp.server.IServer#isStarted()
     */
    public boolean isStarted() {
        return startFlag;
    }

    /**
     * @see com.webex.mdp.server.IServer#start()
     * @throws Exception
     *             error
     */
    public void start() throws Exception {
        if (startFlag) {
            return;
        }

        this.notificationBus.start();
        this.notificationBus.getChannel().setOpt(Channel.LOCAL, Boolean.FALSE);

        exector.execute(this);
        if (logger.isDebugEnabled()) {
            logger.debug("groupManager started");
        }
        startFlag = true;
    }

    /**
     * @see com.webex.mdp.server.IServer#stop()
     * @throws Exception
     *             error
     */
    public void stop() throws Exception {
        if (!startFlag) {
            return;
        }
        if (logger.isDebugEnabled()) {
            logger.debug("--- stop --- ");
        }
        this.stopRequest = true;

     
        // actually, call channel.close
        notificationBus.stop();
        exector.shutdownNow();
             startFlag = false;

    }
    

    /**
     * Gets the local address.
     * 
     * @return the local address
     */
    public Address getLocalAddress() {
        return notificationBus.getLocalAddress();
    }

    /**
     * Gets the membership.
     * 
     * @return the membership
     */
    public Vector<Address> getMembership() {
        return notificationBus.getMembership();
    }

    /**
     * @see java.lang.Runnable#run()
     */
    public void run() {
        String msg = "current time is " + new Date();
        while (!stopRequest) {
            try {
                sendNotification(msg);
                logger.info("Members: " + this.getMembership());
                Thread.sleep(MSG_INTERVAL_MS);
            } catch(InterruptedException e) {
                logger.error("Interrupted");
            } catch (Exception e) {
                logger.error(e);
            }
        }

    }

    
    /**
     * @return String
     */
    public String toString() {
        StringBuilder sb = new StringBuilder(this.myName);
        if (null != this.myChannel) {
            sb.append("\nChannel: " + this.myChannel.getProperties());
        }
        return sb.toString();
    }

    
    public void setConsumer(NotificationConsumer consumer) {
        this.consumer = consumer;
    }

    public static void main(String[] args) throws Exception {
        String groupName = "MdpGroup";
        String host = "224.1.2.3";
        int port = 1975;
        if(args.length >= 2) {
            host = args[0];
            port = NumberUtils.toInt(args[1]);
            if(args.length > 2) {
                groupName = args[2];
            }
        }
        
        NotificationConsumer consumer = new NotificationConsumer();
        final NotificationManager mgr = new NotificationManager(host,port);
        mgr.setName(groupName);
        mgr.setConsumer(consumer);
        mgr.init();
        mgr.start();
        System.out.println(String.format("started on %s:%d as %s", host, port, groupName ));
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                System.out.println("Byebye");
            }
        }));
        Signal.handle(new Signal("INT"), new SignalHandler () {
            public void handle(Signal sig) {
                System.out.println("Got a signal--" + sig);
                try {
                    mgr.stop();
                    mgr.clean();
                } catch (Exception e) {
                   logger.error(e);
                }

            }
          });
          while(mgr.isStarted()) {
            Thread.sleep(2000);
            }
       
    }
}
